/*******************************************************************************
 * Copyright 2011 Alexandre Zglav and Morphiastic
 * 
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 * 
 *   http://www.apache.org/licenses/LICENSE-2.0
 * 
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 ******************************************************************************/
package com.morphiastic.dao;

import java.io.IOException;
import java.util.LinkedList;
import java.util.List;

import org.elasticsearch.action.search.SearchRequestBuilder;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.action.search.SearchType;
import org.elasticsearch.client.Client;
import org.elasticsearch.common.unit.TimeValue;
import org.elasticsearch.index.query.QueryBuilder;
import org.elasticsearch.search.SearchHit;

import com.google.code.morphia.Key;
import com.google.code.morphia.dao.BasicDAO;
import com.google.code.morphia.query.Query;
import com.google.code.morphia.query.UpdateOperations;
import com.google.code.morphia.query.UpdateResults;
import com.mongodb.WriteConcern;
import com.mongodb.WriteResult;
import com.morphiastic.MorphiasticDataStore;
import com.morphiastic.conversion.ElasticConverter;

public class MorphiasticBasicDAO<T, K> extends BasicDAO<T, K> implements MorphiasticDAO<T, K>{

	
	MorphiasticDataStore morphiasticDataStore;
//	IndexedClass indexedClass;
	
	private ElasticConverter<K> idConverter;
	
	@SuppressWarnings("unchecked")
	public MorphiasticBasicDAO(Class<T> entityClass, MorphiasticDataStore ds) {
		super(entityClass, ds);
		this.morphiasticDataStore = ds;
		ds.addIndexedClass(entityClass);
		this.idConverter = ds.getIdConverter(entityClass);
	}

	
	@Override
	public List<T> getAll() {
		 return createQuery().asList();
	}

	
	public List<T> getAll(List<K> keys){
		if(keys == null || keys.isEmpty()) return null; //todo check this ... 
		System.out.println("getall keys = " + keys);
		Query<T> query =  createQuery().field("_id").in(keys);
		System.out.println("getAllquery : "+query);
		return query.asList();
//		return ds.f.disableValidation().filter("_id in", keys).asList();
	}
	
 
	@Override
	public SearchResponse search(QueryBuilder q, int size) {
		return search(q, 0, size, true);
	}


	@Override
	public SearchResponse search(QueryBuilder q, int from, int size) {
		 return search(q, from, size, true);
	}

	@Override
	public SearchResponse search(QueryBuilder q, int from, int size,  boolean includeDescendants) {
		 return search(q, from, size, SearchType.DEFAULT, includeDescendants);
	}
	

	@Override
	public SearchResponse search(QueryBuilder q, int from, int size,
			SearchType searchType, boolean includeDescendants) {
	
		String[] types = getInvolvedTypes(includeDescendants);
		
		
		System.out.println("INVOLVED TYPES = "  );
		for (String string : types) {
			System.out.println(string);
		}
//		TODO we should also make sure that the query is not fired against all types if types[] is null or empty ...
//		What should we do ? return null ? ... 
//				
//		
		
		SearchRequestBuilder builder = getElasticClient()
		.prepareSearch(getGlobalElasticIndexname())
        .setQuery(q)
        .setFrom(from)
        .setSize(size)
        .setTypes(types)
        .setSearchType(searchType)
        .setExplain(true); //TODO set to false here for prod
		
		
		
		
//		System.out.println("SEARCH this.indexedClass.getTypeName() : " + types);
		System.out.println( "search request : \n " + builder.toString());
		System.out.println("SENDING SEARCH ...");
		
		
		SearchResponse response =  builder.execute()
        .actionGet();
		
		return response;
		
	}

	
	
	public SearchResponse search(SearchRequestBuilder searchRequestBuilder, boolean includeDescendants){
		String[] types = getInvolvedTypes(includeDescendants);
		searchRequestBuilder.setTypes(types);
		return searchRequestBuilder.execute().actionGet();
	}
	
	
	@Override
	public List<T> searchAndFetch(QueryBuilder q, int from, int size, SearchType searchType, boolean includeDescendants){
//		System.out.println("Search and fetch triggered");
		//TODO manage polymorphism here
		
		String[] types = getInvolvedTypes(includeDescendants);
		SearchRequestBuilder builder = getElasticClient()
		.prepareSearch(getGlobalElasticIndexname())
        .setQuery(q)
        .setFrom(from)
        .setSize(size)
        .setTypes(types)
        .addField("_mid")
        .setSearchType(searchType)
        .setExplain(true); 
		
		SearchResponse response =  builder.execute()
        .actionGet();
		LinkedList<K> ids = new LinkedList<K>();
		for (SearchHit hit : response.getHits()) {
			ids.add(this.idConverter.fromIndexValue(hit.field("_mid").value()));
		}		

//		System.out.println("found ids ");
//		System.out.println(ids);

		List<T> all = getAll((List<K>) ids);
		
		return all;
		
		
//		return null;
	}
	

	@Override
	public SearchResponse searchScan(QueryBuilder q, int size) {
		 return searchScan(q, size, 60000, true);
	}


	@Override
	public SearchResponse searchScan(QueryBuilder q, int size,
			long scrollTimeMillis, boolean includeDescendants) {
		
		String[] types = getInvolvedTypes(includeDescendants);
		
		SearchResponse scrollResp = getElasticClient()
		.prepareSearch(getGlobalElasticIndexname())
		.setTypes(types)
        .setSearchType(SearchType.SCAN)
        .setScroll(new TimeValue(scrollTimeMillis))
        .setQuery(q.buildAsBytes().copiedByteArray())
        .setSize(size).execute().actionGet();

		return scrollResp;
	}

	
	@Override
	public Client getElasticClient() {
		return this.morphiasticDataStore.getElasticClient();
	}
	
	@Override
	public String getGlobalElasticIndexname() {
		return this.morphiasticDataStore.getElasticStore().getGlobalIndexName();
	}
	
	@Override
	public Key<T> save(T entity) {
		
		System.out.println("MorphiasticBasicDAO save : entity class = " + entity.getClass() );
		
		checkEntityClassIsKnown(entity);
		Key<T> key = super.save(entity);
		//create a JSON object 
//		this.morphiasticDataStore.indexEntity(entityClazz, entity);  //changed on 10.06.2012 because it would create problems when a subclass was saved using a parent's DAO ... the type mapping would not fit 
		this.morphiasticDataStore.indexEntity( entity.getClass() , entity);  //this is still in test and should be impacted on other methods as well ... 
		return key;
	}
	
	@Override
	public Key<T> save(T entity, WriteConcern wc) {
		checkEntityClassIsKnown(entity);
		Key<T> key = super.save(entity, wc);
//		this.morphiasticDataStore.indexEntity(entityClazz, entity);//changed on 10.06.2012
		this.morphiasticDataStore.indexEntity( entity.getClass(), entity);
		return key;
	}
	
	
	@Override
	public void index(T entity) {
		checkEntityClassIsKnown(entity);
//		this.morphiasticDataStore.indexEntity(entityClazz, entity); //changed on 10.06.2012
		this.morphiasticDataStore.indexEntity( entity.getClass(), entity);
	};
	
	
	
	/**
	 * IMPORTANT NOTE : as of now it is impossible to update the ElasticSearch index if an update is performed using this method 
	 * 
	 * 
	 * (non-Javadoc)
	 * @see com.google.code.morphia.DAO#updateFirst(com.google.code.morphia.query.Query, com.google.code.morphia.query.UpdateOperations)
	 */
	public UpdateResults<T> updateFirst(Query<T> q, UpdateOperations<T> ops) {
		checkEntityClassIsKnown(q.getEntityClass());
		UpdateResults<T> ur = super.updateFirst(q, ops);
		
			//we need to get our hands on the full mongoDB document
//			we need to pass the actual Query to ElasticStore that will execute it and pull the mongo docs to forward them 
//			this.morphiasticDataStore.indexUpdateFirst(entityClazz, q);
			this.morphiasticDataStore.indexUpdateFirst(q.getEntityClass(), q); //changed on 10.06.2012
		 
		return ur;
	}
	
	/* (non-Javadoc)
	 * @see com.google.code.morphia.DAO#update(com.google.code.morphia.query.Query, com.google.code.morphia.query.UpdateOperations)
	 */
	public UpdateResults<T> update(Query<T> q, UpdateOperations<T> ops) {
		checkEntityClassIsKnown(q.getEntityClass());
		UpdateResults<T> ur = super.update(q, ops);
//		this.morphiasticDataStore.indexUpdate(entityClazz, q);
		this.morphiasticDataStore.indexUpdate(q.getEntityClass(), q);
		return ur;
	}
	
	/* (non-Javadoc)
	 * @see com.google.code.morphia.DAO#delete(T)
	 */
	public WriteResult delete(T entity) {
		checkEntityClassIsKnown(entity);
		WriteResult wr = super.delete(entity);
//		this.morphiasticDataStore.deleteFromIndex(entityClazz, entity);
		this.morphiasticDataStore.deleteFromIndex(entity.getClass(), entity);
		return wr;
	}

	/* (non-Javadoc)
	 * @see com.google.code.morphia.DAO#delete(T, com.mongodb.WriteConcern)
	 */
	public WriteResult delete(T entity, WriteConcern wc) {
		checkEntityClassIsKnown(entity);
		WriteResult wr = super.delete(entity, wc);
//		this.morphiasticDataStore.deleteFromIndex(entityClazz, entity);
		this.morphiasticDataStore.deleteFromIndex(entity.getClass(), entity);
		return wr;
	}
	 
	
	
	@Override
	public WriteResult deleteByQuery(Query q) {
		checkEntityClassIsKnown(q.getEntityClass());
//		this.morphiasticDataStore.deleteFromIndexByQuery(entityClazz,q);
		this.morphiasticDataStore.deleteFromIndexByQuery(q.getEntityClass(),q);
		return super.deleteByQuery(q);
	}
	
	
	
	private String[] getInvolvedTypes(boolean includeDescendants){
		String[] types = null;
		if(includeDescendants){
			types = morphiasticDataStore.getEntityAssociatedTypeNames(super.entityClazz);
		}else{
			String typeName = morphiasticDataStore.getEntityTypeName(super.entityClazz);
			if(typeName!=null){
				types = new String[1];
				types[0] = typeName;
			}
		}
		return types;
	}
	
	
	/**
	 * //just a litle check that the entity we are trying to save has properly been analyzed
		//this may happen if we instatiate a superclass dao and try to save a children entity
		//with it. The problem is that ES doesnt yet know about the type so we might have to create mappings for it
	 * Just makes sure that the class that we are trying to save is already known
	 * otherwise it will try to create propper mapping on ES
	 * @param entity
	 */
	private void checkEntityClassIsKnown(T entity){
		Class<?> c = entity.getClass();
		checkEntityClassIsKnown(c);
	}
	
	private void checkEntityClassIsKnown(Class<?> clazz){
		morphiasticDataStore.getElasticStore().addIndexedClass(clazz);
	}


	 

	@Override
	public Key<T> saveOnly(T entity) {
		return super.save(entity);
	}


	@Override
	public UpdateResults<T> updateOnly(Query<T> q, UpdateOperations<T> ops) {
		return  super.update(q, ops);
	}

	
	 
	
	
}
